package com.atguigu.gmall.realtime.app.marketing.func;

import com.atguigu.gmall.realtime.app.marketing.bean.MarketingMessage;
import com.atguigu.gmall.realtime.app.marketing.constant.MarketingConst;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class PushLimiter extends KeyedProcessFunction<Tuple2<String,String>, MarketingMessage,MarketingMessage> {


    ValueState<String>  pushLockState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("push_lock", String.class);
        pushLockState=getRuntimeContext().getState(stateDescriptor);
    }

    @Override
    public void processElement(MarketingMessage marketingMessage, KeyedProcessFunction<Tuple2<String, String>, MarketingMessage, MarketingMessage>.Context ctx, Collector<MarketingMessage> out) throws Exception {
        String pushLock = pushLockState.value();
        if(pushLock==null){
            out.collect(marketingMessage);
            //加锁
            pushLockState.update("1");
            Integer limitTimeNum = marketingMessage.getLimitTimeNum();
            String limitTimeUnit = marketingMessage.getLimitTimeUnit();
            Integer expireMs = getMillSec(limitTimeNum, limitTimeUnit);
            long curMs = ctx.timerService().currentProcessingTime();
            ctx.timerService().registerProcessingTimeTimer(curMs+expireMs);

        }else{
            System.out.println("该消息被拦截："+marketingMessage);
        }
    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Tuple2<String, String>, MarketingMessage, MarketingMessage>.OnTimerContext ctx, Collector<MarketingMessage> out) throws Exception {
        pushLockState.clear(); //达到过期时间清除锁
    }

    private Integer getMillSec(Integer timeNum, String timeUnit  ) {
        switch (timeUnit){
            case MarketingConst.TIME_UNIT_DAY: return timeNum*3600*24*1000 ;
            case MarketingConst.TIME_UNIT_HOUR: return timeNum*3600*1000;
            case MarketingConst.TIME_UNIT_MINIUTE: return timeNum*60*1000;
            case MarketingConst.TIME_UNIT_SEC: return timeNum*1000 ;
        }
        return timeNum;
    }


}
